package kuoge.practice.kafka.stream.member.rfm;

import com.alibaba.fastjson.JSON;
import io.netty.util.HashedWheelTimer;
import io.netty.util.TimerTask;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import kuoge.practice.kafka.stream.member.KafkaProducerFactory;

import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * @Description
 */
public class OrderCompletedEventProducer {
    private static final Producer<String, String> kafkaProducer = KafkaProducerFactory.getInstance();
    private static final String ORDER_TOPIC = "order_topic";
    private final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();

    void startSendEvent() {
        user_a_event();

        user_b_event();

        System.out.println("start send event success");
    }

    private void user_a_event() {
        TimerTask task = timeout -> {
            final long amountPrice = ThreadLocalRandom.current().nextLong(10, 300);
            final OrderCompletedEvent orderCompletedEvent = new OrderCompletedEvent("member_a", amountPrice);
            kafkaProducer.send(new ProducerRecord<>(ORDER_TOPIC, orderCompletedEvent.getMemberId() + "_" + System.currentTimeMillis(), JSON.toJSONString(orderCompletedEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            System.out.println("send member_a event, thread:" + Thread.currentThread().getName());
            user_a_event();
        };
        hashedWheelTimer.newTimeout(task, 1000, TimeUnit.MILLISECONDS);
    }

    private void user_b_event() {
        TimerTask task = timeout -> {
            final long amountPrice = ThreadLocalRandom.current().nextLong(100, 600);
            final OrderCompletedEvent orderCompletedEvent = new OrderCompletedEvent("member_b", amountPrice);
            kafkaProducer.send(new ProducerRecord<>(ORDER_TOPIC, orderCompletedEvent.getMemberId() + "_" + System.currentTimeMillis(), JSON.toJSONString(orderCompletedEvent)), (rs, ex) -> System.out.println(rs.offset() + ", " + ex));
            System.out.println("send member_b event, thread:" + Thread.currentThread().getName());
            user_b_event();
        };
        hashedWheelTimer.newTimeout(task, 5000, TimeUnit.MILLISECONDS);
    }

    void stopSendEvent() {
        hashedWheelTimer.stop();
    }
}
